package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2DataSource {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("source")
      .master("local")
      .getOrCreate()

    /**
      *
      * 读取json格式的数据
      * json 中自带列名和列的类型的，spark 在读取的过程中会自动解析
      *
      */

    spark
      .read
      .format("json")
      .load("data/students.json")
      .show()

    /**
      * 读取csv格式的数据
      *
      */
    val csvDF: DataFrame = spark
      .read
      .format("csv")
      //列分隔方式，默认是逗号
      .option("sep", ",")
      //增加表结构
      .schema("id STRING,name STRING ,age INT , gender STRING,clazz STRING")
      .load("data/students.txt")

    csvDF.show()
    csvDF.printSchema()

    /**
      * 保存为parquet格式
      *
      */

    //csvDF.write.mode(SaveMode.Overwrite).parquet("data/parquet")


    /**
      *
      * 读取parquet格式的数据
      * 1、自带列名
      * 2、数据会压缩 (压缩和解压需要时间，压缩比一般在5倍左右)
      * 3、可以兼容hive
      */


    val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("data/parquet")


    parquetDF.printSchema()
    parquetDF.show()


    //csvDF.write.mode(SaveMode.Overwrite).orc("data/orc")

    /**
      * 读取orc格式
      * 1、自带列名
      * 2、数据会压缩 (压缩和解压需要时间，压缩比一般在5倍左右)
      * 3、可以兼容hive
      */


    val orcDF: DataFrame = spark
      .read
      .format("orc")
      .load("data/orc")

    orcDF.show()
    orcDF.printSchema()


    /**
      * 读取jdbc中的数据
      *
      */


    val jdbcDF: DataFrame = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "student.student")
      .option("user", "root")
      .option("password", "123456")
      .load()

    jdbcDF.show()


    /**
      * 将数据保存到mysql
      *
      */

    orcDF.write
      .format("jdbc")
      .mode(SaveMode.Overwrite)
      .option("url", "jdbc:mysql://master:3306?useUnicode=true&characterEncoding=utf-8")
      .option("dbtable", "student.student1")
      .option("user", "root")
      .option("password", "123456")
      //手动指定列的类型
      .option("createTableColumnTypes", "name varchar(64), id  varchar(64),age int,gender  varchar(64),clazz  varchar(64)")
      .save()

  }

}
